package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo9ReduceByKey {
  def main(args: Array[String]): Unit = {
    // 统计班级人数
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo8GroupBy")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    // 1、读取students数据
    val linesRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt")

    // 2、将数据构建成 (班级,1) K-V格式

    val clazzRDD: RDD[(String, Int)] = linesRDD.map(line => (line.split(",")(4), 1))

    clazzRDD.groupByKey().map(kv => (kv._1, kv._2.sum)).foreach(println)

    /**
     * reduceByKey：转换算子、分区类算子（只能作用在K-V格式的RDD上）
     * 相比较于groupByKey：性能更高但功能较弱
     * 需要接收一个 聚合函数f，一般是 加和、最大值、最小值
     * 相当于MR中的combiner
     * 会在Map进行预聚合
     * 只适用于幂等操作
     * y=f(x)=f(y)=f(f(x))
     */
    clazzRDD.reduceByKey((i: Int, j: Int) => {
      i + j
    }).foreach(println)

    // 简写
    clazzRDD.reduceByKey(_ + _).foreach(println)

    while (true) {

    }
  }

}
